import pandas as pd
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import create_engine, text


# 创建数据库连接
def get_engine(user, password, host, port, database):
    return create_engine(f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}")

# 从不同的数据源读取数据
def read_data_from_source_1(engine):
    query = "SELECT segment_id, coverage, deal_status, report_dt FROM segment_csp_byday WHERE DATE(report_dt) = CURDATE() AND (report_dt, segment_id) IN (SELECT MAX(report_dt) AS report_dt, segment_id FROM segment_csp_byday WHERE DATE(report_dt) = CURDATE() GROUP BY segment_id)"
    return pd.read_sql(query, engine)

def read_data_from_source_2(engine):
    query = "SELECT segment_id, sum FROM segment_push_notice WHERE DATE(create_time) = CURDATE()"
    return pd.read_sql(query, engine)

# 更新 segment_csp_byday 表中的 deal_status
def update_deal_status(engine, segment_ids_dates):
    if not segment_ids_dates.empty:
        placeholders = ','.join([f"(:segment_id_{i}, :report_dt_{i})" for i in range(len(segment_ids_dates))])
        query = f"""
        UPDATE segment_csp_byday 
        SET deal_status = 0 
        WHERE (segment_id, report_dt) IN ({placeholders})
        AND DATE(report_dt) = CURDATE()
        """
        params = {}
        for i, row in segment_ids_dates.iterrows():
            params[f"segment_id_{i}"] = row['segment_id']
            params[f"report_dt_{i}"] = row['report_dt']

        try:
            with engine.connect() as connection:
                transaction = connection.begin()
                result = connection.execute(text(query), params)
                if result.rowcount > 0:
                    transaction.commit()
                else:
                    transaction.rollback()
        except SQLAlchemyError as e:
            print(f"Error occurred during update: {e}")
            transaction.rollback()

# 主函数
def main():
    # 数据库连接信息
    user_1, password_1, host_1, port_1, database_1 = 'root', 'd1o0N1z!rbELRf3r', 'rm-2vc9b10eck3rpbbv7.mysql.cn-chengdu.rds.aliyuncs.com', 3306, 'object_modeling_v2'
    user_2, password_2, host_2, port_2, database_2 = 'worthservice', 'y5eG9XoxxeD3THPm', 'rm-2vczz3ais6fi0y7j6.mysql.cn-chengdu.rds.aliyuncs.com', 3306, 'worth-service'

    # 创建引擎
    engine_1 = get_engine(user_1, password_1, host_1, port_1, database_1)
    engine_2 = get_engine(user_2, password_2, host_2, port_2, database_2)

    try:
        # 读取数据
        df_byday = read_data_from_source_1(engine_1)
        df_segment_push_notice = read_data_from_source_2(engine_2)
    except SQLAlchemyError as e:
        print(f"Error reading data: {e}")
        return
    # 打印读取的数据以进行调试
    print("Data from segment_csp_byday:")
    print(df_byday)
    print("\nData from segment_push_notice:")
    print(df_segment_push_notice)

    # 合并数据并找到不匹配的行
    merged_df = pd.merge(df_byday, df_segment_push_notice, on='segment_id', how='left')
    print("\nMerged Data:")
    print(merged_df)
    mismatched_segments = merged_df[(merged_df['deal_status'] == 1) & ((merged_df['sum'].isna()) | (merged_df['coverage'] != merged_df['sum']))][['segment_id', 'report_dt']]
    print("\nMismatched segments:")
    print(mismatched_segments)
    # 确保 mismatched_segments 是 DataFrame
    mismatched_segments = mismatched_segments.reset_index(drop=True)

    if not mismatched_segments.empty:
        # 更新数据库
        try:
            update_deal_status(engine_1, mismatched_segments)
        except SQLAlchemyError as e:
            print(f"Error updating data: {e}")

    else:
        print("No mismatched segments found.")

# 运行主函数
if __name__ == "__main__":
    main()
